package com.atguigu.edu.realtime.app.dwd.db;

import com.atguigu.edu.realtime.util.MyKafkaUtil;
import com.atguigu.edu.realtime.util.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Author: Felix
 * Date: 2022/5/11
 * Desc: 交易域订单预处理表
 */
public class DwdTradeOrderPreProcess {
    public static void main(String[] args) throws Exception {

        // TODO 1. 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2. 检查点相关设置(略)

        // TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        tableEnv.executeSql("create table topic_db(" +
                "`database` String, " +
                "`table` String, " +
                "`type` String, " +
                "`data` map<String, String>, " +
                "`old` map<String, String>, " +
                "`proc_time` as PROCTIME(), " +
                "`ts` string " +
                ")" + MyKafkaUtil.getKafkaDDL("topic_db", "dwd_trade_order_pre_process"));




        // TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        tableEnv.executeSql("create table dwd_trade_order_source(" +
                "`common` map<String, String>, " +
                "`page` map<String, String>, " +
                "`proc_time` as PROCTIME(), " +
                "`ts` string " +
                ")" + MyKafkaUtil.getKafkaDDL("dwd_trade_order_source", "dwd_trade_order_pre_process"));



        Table sourceInfo = tableEnv.sqlQuery("select  " +
                "common['sc'] source_id, " +
                "page['item'] order_id, " +
                "ts si_ts, " +
                "proc_time " +
                "from `dwd_trade_order_source` ");
        tableEnv.createTemporaryView("source_info", sourceInfo);




        // TODO 4. 读取订单明细表数据
        Table orderDetail = tableEnv.sqlQuery("select  " +
                "data['id'] id, " +
                "data['order_id'] order_id, " +
                "data['course_id'] course_id, " +
                "data['course_name'] course_name, " +
                "data['create_time'] create_time, " +
                "cast(cast(data['origin_amount'] as decimal(16,2)) as String) original_amount, " +
                "data['coupon_reduce'] coupon_reduce, " +
                "data['final_amount'] final_amount, " +
                "ts od_ts, " +
                "proc_time " +
                "from `topic_db` where `table` = 'order_detail' " +
                "and `type` = 'insert' ");
        tableEnv.createTemporaryView("order_detail", orderDetail);




        // TODO 5. 读取订单表数据
        Table orderInfo = tableEnv.sqlQuery("select  " +
                "data['id'] id, " +
                "data['user_id'] user_id, " +
                "data['province_id'] province_id, " +
                "data['create_time'] create_time, " +
                "data['update_time'] update_time, " +
                "data['order_status'] order_status, " +
                "`type`, " +
                "`old`, " +
                "ts oi_ts " +
                "from `topic_db` " +
                "where `table` = 'order_info' " +
                "and (`type` = 'insert' or `type` = 'update')");
        tableEnv.createTemporaryView("order_info", orderInfo);




        // TODO 9. 关联五张表获得订单明细表
        Table resultTable = tableEnv.sqlQuery("select  " +

                "od.id, " +
                "od.order_id, " +
                "od.course_id, " +
                "od.course_name, " +
                "date_format(od.create_time, 'yyyy-MM-dd') date_id, " +
                "od.create_time, " +
                "od.original_amount, " +
                "od.coupon_reduce, " +
                "od.final_amount, " +
                "od.od_ts, " +
                "oi.user_id, " +
                "oi.province_id, " +
                "date_format(oi.create_time, 'yyyy-MM-dd') operate_date_id, " +
                "oi.create_time, " +
                "oi.update_time, " +
                "oi.order_status, " +
                "oi.`type`, " +
                "oi.`old`, " +
                "oi.oi_ts, " +
                "si.source_id, "+
                "current_row_timestamp() row_op_ts " +
                "from order_detail od  " +
                " join order_info oi " +
                "on od.order_id = oi.id " +
                "left join source_info si "+
                "on od.order_id = si.order_id"
                );
        tableEnv.createTemporaryView("result_table", resultTable);



        // TODO 10. 建立 Upsert-Kafka dwd_trade_order_pre_process 表
        tableEnv.executeSql("" +
                "create table dwd_trade_order_pre_process( " +
                "id string, " +
                "order_id string, " +
                "course_id string, " +
                "course_name string, " +
                "date_id string, " +
                "create_time string, " +
                "original_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "od_ts string, " +
                "user_id string, " +
                "province_id string, " +
                "operate_date_id string, " +
                "create_time0 string, " +
                "update_time string, " +
                "order_status string, " +
                "`type` string, " +
                "`old` map<string,string>, " +
                "oi_ts string, " +
                "source_id string, " +
                "row_op_ts timestamp_ltz(3), " +
                "primary key(id) not enforced " +
                ")" + MyKafkaUtil.getUpsertKafkaDDL("dwd_trade_order_pre_process"));

        // TODO 11. 将关联结果写入 Upsert-Kafka 表
        tableEnv.executeSql("" +
                "insert into dwd_trade_order_pre_process  " +
                "select * from result_table");


    }
}
